package cn.doitedu.rtmk.demo11;

import cn.doitedu.rtmk.beans.RuleMetaBean;
import cn.doitedu.rtmk.beans.UserEvent;
import cn.doitedu.rtmk.interfaces.IRuleCalculatorV4;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import groovy.lang.GroovyClassLoader;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.RandomUtils;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.roaringbitmap.RoaringBitmap;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
public class RuleEngineCoreFunction11 extends KeyedBroadcastProcessFunction<Long, UserEvent, RuleMetaBean, JSONObject> {

    // 本Map对象有可能同时被processElement方法和processBroadcastElement方法同时操作，
    // 一个读，一个写，有可能发生线程安全问题，所以采用并发安全的 ConcurrentHashMap
    Map<Integer, IRuleCalculatorV4> ruleCalculatorMap = new ConcurrentHashMap<>();

    MapState<Integer, Integer> calculator_status;

    ListState<UserEvent> eventListState;


    @Override
    public void open(Configuration parameters) throws Exception {

        // 缓存最近1分钟的用户行为数据
        StateTtlConfig ttlConfig = StateTtlConfig
                .newBuilder(Time.minutes(1))
                .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
                .build();

        ListStateDescriptor<UserEvent> listStateDescriptor = new ListStateDescriptor<>("events", UserEvent.class);
        listStateDescriptor.enableTimeToLive(ttlConfig);
        eventListState = getRuntimeContext().getListState(listStateDescriptor);

        // 用于记录规则运算机是否是第一次运行
        calculator_status = getRuntimeContext().getMapState(new MapStateDescriptor<Integer, Integer>("calculator_status", Integer.class, Integer.class));

    }

    /**
     * 处理主流数据的方法
     * 静态圈选条件：月平均消费额>200的用户
     * 触发条件：当他发生A行为时（event_id='page_load', properties['url'] = '/page/001'），命中规则
     */
    @Override
    public void processElement(UserEvent userEvent, KeyedBroadcastProcessFunction<Long, UserEvent, RuleMetaBean, JSONObject>.ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {

        /*if(userEvent.getEvent_id().equals("x") && RandomUtils.nextInt(1,11)%3==0){
            throw new RuntimeException("整死你..............");
        }*/

        ReadOnlyBroadcastState<Integer, RuleMetaBean> broadcastState = ctx.getBroadcastState(new MapStateDescriptor<Integer, RuleMetaBean>("bcRuleBean", Integer.class, RuleMetaBean.class));

        // 如果 运算机池 为空，则说明系统可能是failover重启过了,先进行重建
        if (ruleCalculatorMap.isEmpty()) {
            restoreCalculatorMap(broadcastState);
        }

        // 遍历运算机集合中的每一个规则运算机
        for (Map.Entry<Integer, IRuleCalculatorV4> entry : ruleCalculatorMap.entrySet()) {
            Integer ruleId = entry.getKey();
            IRuleCalculatorV4 ruleCalculator = entry.getValue();

            // 判断该运算机是否是第一次就位执行，
            Integer status = calculator_status.get(ruleId);

            // 如果是，则需要把缓存的最近1分钟的数据交给它处理
            if (status == null) {
                for (UserEvent stateEvent : eventListState.get()) {
                    ruleCalculator.calc(stateEvent, out);
                }
                calculator_status.put(ruleId, 1);
            }
            // 然后再处理最新的这一条
            ruleCalculator.calc(userEvent, out);
        }

        // 将当前的数据存入缓存
        eventListState.add(userEvent);

    }


    /**
     * 处理广播流数据的方法
     */
    @Override
    public void processBroadcastElement(RuleMetaBean ruleMetaBean, KeyedBroadcastProcessFunction<Long, UserEvent, RuleMetaBean, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {

        // 获取广播状态
        BroadcastState<Integer, RuleMetaBean> broadcastState =
                ctx.getBroadcastState(new MapStateDescriptor<Integer, RuleMetaBean>("bcRuleBean", Integer.class, RuleMetaBean.class));


        // 如果 运算机池 为空，则说明系统可能是failover重启过了
        if (ruleCalculatorMap.isEmpty()) {
            restoreCalculatorMap(broadcastState);
        }

        // 取出新收到的规则元信息中的： 规则id 和 预圈选人群bytes
        int rule_id = (int) ruleMetaBean.getRule_id();

        /**
         * 此处是规则下线、停用功能处理
         * rowKind ==>  0: +I  , 1:-U  ,2: +U  , 3: -D
         */
        int rowKind = ruleMetaBean.getRowKind();
        // 规则元数据表中的 规则删除和 规则停用，在我们后端的规则引擎中，都代表将运算机移除
        if (rowKind == 3 || ruleMetaBean.getOnline_status() == 0) {
            ruleCalculatorMap.remove(rule_id);
            log.warn("移除了一个规则:{}", rule_id);
        } else {

            // 一、将接收到的元数据信息Bean，存入广播状态
            broadcastState.put(rule_id, ruleMetaBean);


            // 二、根据本次收到的元数据信息Bean，构建运算机
            IRuleCalculatorV4 calculator = buildRuleCalculator(ruleMetaBean, rule_id);


            // 三、将构建好的运算机，放入运算机池
            ruleCalculatorMap.put(rule_id, calculator);
        }
    }


    /**
     * 重建恢复运算机池
     */
    private void restoreCalculatorMap(ReadOnlyBroadcastState<Integer, RuleMetaBean> broadcastState) throws Exception {
        // 遍历广播状态中存储的元数据Bean

        Iterable<Map.Entry<Integer, RuleMetaBean>> entries = broadcastState.immutableEntries();
        for (Map.Entry<Integer, RuleMetaBean> entry : entries) {
            Integer ruleId = entry.getKey(); // 规则id
            RuleMetaBean ruleMetaBean = entry.getValue(); // 元数据bean

            // 根据元数据bean，构建运算机
            IRuleCalculatorV4 calculator = buildRuleCalculator(ruleMetaBean, ruleId);
            log.warn("故障后重启,恢复了一个运算机,ruleId:{}",ruleId);
            // 将构建的运算机，放入运算机池
            ruleCalculatorMap.put(ruleId, calculator);
        }
    }



    private IRuleCalculatorV4 buildRuleCalculator(RuleMetaBean ruleMetaBean, int rule_id) throws IOException, InstantiationException, IllegalAccessException {
        // 1.反序列化出预圈选人群的bitmap对象
        byte[] pre_select_users = ruleMetaBean.getPre_select_users();
        RoaringBitmap preSelectUsersBitmap = RoaringBitmap.bitmapOf();
        ByteBuffer byteBuffer = ByteBuffer.wrap(pre_select_users);
        preSelectUsersBitmap.deserialize(byteBuffer);

        // 2.提取规则模型id
        int rule_model_id = ruleMetaBean.getRule_model_id();

        // 3.提取规则的条件参数json
        String rule_param_json = ruleMetaBean.getRule_param_json();


        /**
         * 4.构建运算机对象
         */
        // 4.1 提取规则所属模型的运算机源代码
        String rule_model_calc_groovy_code = ruleMetaBean.getRule_model_calc_groovy_code();
        // 4.2 利用groovy类加载器，对源代码进行编译加载
        GroovyClassLoader groovyClassLoader = new GroovyClassLoader();
        // 4.3 反射出运算机的示例对象
        Class aClass = groovyClassLoader.parseClass(rule_model_calc_groovy_code);
        IRuleCalculatorV4 calculator = (IRuleCalculatorV4) aClass.newInstance();
        // 初始化运算机
        calculator.open(preSelectUsersBitmap,
                JSON.parseObject(rule_param_json),
                getRuntimeContext(),
                ruleMetaBean.getInit_value_endtime());
        //log.warn("收到一条新规则的元信息, 规则id:{},所属模型id:{},规则预圈选人群人数:{},规则参数:{}", rule_id, rule_model_id, preSelectUsersBitmap.getCardinality(), rule_param_json);

        return calculator;
    }

}
